package com.atguigu.api2

import com.atguigu.api.SensorReading
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @description: xxx
 * @time: 2020/7/3 18:06
 * @author: baojinlong
 **/
object ProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    val inputStream: DataStream[String] = environment.socketTextStream("localhost", 7777)
    val dataStream: DataStream[SensorReading] = inputStream.map(data => {
      val dataArray: Array[String] = data.split(",")
      SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
    })

    // 检测每一个传感器温度是否连续上升,在10秒之内,只要10s内数据上升了就报警.不用考虑一直上升
    dataStream
      .keyBy("id") // keyBy后都是仅仅根据key来计算的
      .process(new TempIncrementWarning(1000L))

    environment.execute("test process")

  }
}

class TempIncrementWarning(interval: Long) extends KeyedProcessFunction[Tuple, SensorReading, String] {
  // 由于需要跟之前的温度做对比,所以需要将上一个温度保存成状态
  lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  // 为了方便删除定时器,还需要保存定时器的时间戳
  lazy val curTimeTimestampStatus: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("curTime", classOf[Long]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[Tuple, SensorReading, String]#Context, out: Collector[String]): Unit = {
    // 首先取出状态
    val lastTemp: Double = lastTempState.value
    val curTimerTs: Long = curTimeTimestampStatus.value
    // 将上次温度值的状态更新为当前数据的温度值
    if (value.temperature > lastTemp && curTimerTs == 0) {
      val ts: Long = ctx.timerService().currentProcessingTime() + interval
      ctx.timerService().registerEventTimeTimer(ts)
      curTimeTimestampStatus.update(ts)
    } else if (value.temperature < lastTemp) {
      // 如果温度下降就删除定时器
      ctx.timerService().deleteEventTimeTimer(curTimerTs)
      // 清空状态
      curTimeTimestampStatus.clear()
    }
  }

  // 定时器触发了,说明10s内没有来下降的温度值,直接报警
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect("温度值连续" + interval / 1000 + "秒上升")
    // 清空状态
    curTimeTimestampStatus.clear()
  }
}